package org.rency.mq.activemq.listener.failover.impl;

import com.alibaba.fastjson.JSON;
import org.rency.mq.activemq.constant.DeliveryMode;
import org.rency.mq.activemq.constant.DestinationType;
import org.rency.mq.activemq.constant.MessageFormat;
import org.rency.mq.activemq.jms.service.JmsAccessor;
import org.rency.mq.activemq.listener.failover.FailoverService;
import org.rency.mq.activemq.listener.failover.RetryExecutor;
import org.rency.mq.activemq.listener.failover.RetryStrategy;
import org.rency.mq.activemq.request.MQRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.JMSException;
import javax.jms.Session;
import java.util.Map;

public abstract class AbstractRetryStrategy implements RetryStrategy {

    private static Logger logger        = LoggerFactory.getLogger(AbstractRetryStrategy.class);
    private JmsAccessor mqAccessor;
    private RetryExecutor retryExecutor;
    private long          interval   = 120;
    private boolean       isSendToFail = false;

    public void retry(MQRequest request) {
        if (needRetry(request)) {
            beforeProcess(request);

            try {
                request.addProperty("isSendToFail", isSendToFail);
                sendToOriginalQueue(request);
            } catch (Exception e) {
                logger.error("Fail to send message to original queue.", e);
                sendToFailedQueue(request);
            }
        } else {
            sendToFailedQueue(request);
        }
    }

    protected void beforeProcess(MQRequest request) {
        request.addProperty(FailoverService.AMQ_SCHEDULED_DELAY, getInterval() * 1000);
    }

    protected abstract boolean needRetry(MQRequest request);

    protected void sendToOriginalQueue(MQRequest request) throws Exception {
        logger.info("Retry to send MQRequest to original queue: " + request.toString());
        retryExecutor.retrySend(request);
    }

    protected void sendToFailedQueue(MQRequest request) {
        if (!isSendToFail) {
            logger.warn("没有配置发送到失败队列.消息将被丢弃.message:" + request.toString());
            return;
        }
        Map<String, Object> props = request.getProperties();

        sendToFailedQueue((String) props.get("failedQueue"), request);
    }

    private void sendToFailedQueue(String failedQueue, Object request) {
        logger.info("Send to failed queue: " + request.toString());
        try {
            mqAccessor
                .sendMessage(failedQueue, DestinationType.QUEUE, MessageFormat.OBJECT, false,
                    Session.AUTO_ACKNOWLEDGE, DeliveryMode.PERSISTENT,JSON.toJSONString(request));
        } catch (JMSException ex) {
            logger.error("Send message to failed queue failed.", ex);
        }
    }

    public JmsAccessor getMqAccessor() {
        return mqAccessor;
    }

    public void setMqAccessor(JmsAccessor mqAccessor) {
        this.mqAccessor = mqAccessor;
    }

    public void setInterval(long interval) {
        if (interval > 0)
            this.interval = interval;
    }

    public long getInterval() {
        return interval;
    }

    public RetryExecutor getRetryExecutor() {
        return retryExecutor;
    }

    public void setRetryExecutor(RetryExecutor retryExecutor) {
        this.retryExecutor = retryExecutor;
    }

    public boolean isSendToFail() {
        return isSendToFail;
    }

    public void setSendToFail(boolean isSendFail) {
        this.isSendToFail = isSendFail;
    }

}
